{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "32849494",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| default_exp core"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2161d9b5",
   "metadata": {},
   "source": [
    "# Core\n",
    "> The `FastHTML` subclass of `Starlette`, along with the `RouterX` and `RouteX` classes it automatically uses."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "46e2e6e8",
   "metadata": {},
   "source": [
    "This is the source code to fasthtml. You won't need to read this unless you want to understand how things are built behind the scenes, or need full details of a particular API. The notebook is converted to the Python module [fasthtml/core.py](https://github.com/AnswerDotAI/fasthtml/blob/main/fasthtml/core.py) using [nbdev](https://nbdev.fast.ai/)."
   ]
  },
  {
   "cell_type": "markdown",
   "id": "568b20cd",
   "metadata": {},
   "source": [
    "## Imports and utils"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8394fbde",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "import json,uuid,inspect,types,uvicorn\n",
    "from starlette.datastructures import URLPath\n",
    "\n",
    "from fastcore.utils import *\n",
    "from fastcore.xml import *\n",
    "\n",
    "from types import UnionType, SimpleNamespace as ns, GenericAlias\n",
    "from typing import Optional, get_type_hints, get_args, get_origin, Union, Mapping, TypedDict, List, Any\n",
    "from datetime import datetime\n",
    "from dataclasses import dataclass,fields\n",
    "from collections import namedtuple\n",
    "from inspect import isfunction,ismethod,Parameter,get_annotations\n",
    "from functools import wraps, partialmethod, update_wrapper\n",
    "from http import cookies\n",
    "from urllib.parse import urlencode, parse_qs, quote, unquote\n",
    "from copy import copy,deepcopy\n",
    "from warnings import warn\n",
    "from dateutil import parser as dtparse\n",
    "from starlette.requests import HTTPConnection\n",
    "\n",
    "from fasthtml.starlette import *\n",
    "\n",
    "empty = Parameter.empty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7f5d0a72",
   "metadata": {},
   "outputs": [],
   "source": [
    "import time\n",
    "\n",
    "from IPython import display\n",
    "from enum import Enum\n",
    "from pprint import pprint\n",
    "\n",
    "from fastcore.test import *\n",
    "from starlette.testclient import TestClient\n",
    "from starlette.requests import Headers\n",
    "from starlette.datastructures import UploadFile"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "19d3f2a7",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _sig(f): return signature_ex(f, True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "19ffa7e0",
   "metadata": {},
   "source": [
    "We write source code _first_, and then tests come _after_. The tests serve as both a means to confirm that the code works and also serves as working examples. The first declared function, `date`, is an example of this pattern."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e68a76c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def date(s:str):\n",
    "    \"Convert `s` to a datetime\"\n",
    "    return dtparse.parse(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5331a3e7",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "datetime.datetime(2024, 8, 25, 14, 0)"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "date('2pm')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7c820373",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def snake2hyphens(s:str):\n",
    "    \"Convert `s` from snake case to hyphenated and capitalised\"\n",
    "    s = snake2camel(s)\n",
    "    return camel2words(s, '-')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "442a5aac",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Snake-Case'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "snake2hyphens(\"snake_case\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "25f3b8f8",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "htmx_hdrs = dict(\n",
    "    boosted=\"HX-Boosted\",\n",
    "    current_url=\"HX-Current-URL\",\n",
    "    history_restore_request=\"HX-History-Restore-Request\",\n",
    "    prompt=\"HX-Prompt\",\n",
    "    request=\"HX-Request\",\n",
    "    target=\"HX-Target\",\n",
    "    trigger_name=\"HX-Trigger-Name\",\n",
    "    trigger=\"HX-Trigger\")\n",
    "\n",
    "@dataclass\n",
    "class HtmxHeaders:\n",
    "    boosted:str|None=None; current_url:str|None=None; history_restore_request:str|None=None; prompt:str|None=None\n",
    "    request:str|None=None; target:str|None=None; trigger_name:str|None=None; trigger:str|None=None\n",
    "    def __bool__(self): return any(hasattr(self,o) for o in htmx_hdrs)\n",
    "\n",
    "def _get_htmx(h):\n",
    "    res = {k:h.get(v.lower(), None) for k,v in htmx_hdrs.items()}\n",
    "    return HtmxHeaders(**res)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b4b5d95",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_request(url: str='/', headers: dict={}, method: str='get') -> Request:\n",
    "    scope = {\n",
    "        'type': 'http',\n",
    "        'method': method,\n",
    "        'path': url,\n",
    "        'headers': Headers(headers).raw,\n",
    "        'query_string': b'',\n",
    "        'scheme': 'http',\n",
    "        'client': ('127.0.0.1', 8000),\n",
    "        'server': ('127.0.0.1', 8000),\n",
    "    }\n",
    "    receive = lambda: {\"body\": b\"\", \"more_body\": False}\n",
    "    return Request(scope, receive)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36e2cac0",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "HtmxHeaders(boosted=None, current_url=None, history_restore_request=None, prompt=None, request='1', target=None, trigger_name=None, trigger=None)"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "h = test_request(headers=Headers({'HX-Request':'1'}))\n",
    "_get_htmx(h.headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "edbc8d46",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def str2int(s)->int:\n",
    "    \"Convert `s` to an `int`\"\n",
    "    s = s.lower()\n",
    "    if s=='on': return 1\n",
    "    if s=='none': return 0\n",
    "    return 0 if not s else int(s)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e68fead1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1, 0)"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "str2int('1'),str2int('none')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1d53e8e7",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _mk_list(t, v): return [t(o) for o in v]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ef18f76b",
   "metadata": {},
   "source": [
    "## Request and response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5ab74473",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "fh_cfg = AttrDict(indent=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0afb520c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _fix_anno(t):\n",
    "    \"Create appropriate callable type for casting a `str` to type `t` (or first type in `t` if union)\"\n",
    "    origin = get_origin(t)\n",
    "    if origin is Union or origin is UnionType or origin in (list,List):\n",
    "        t = first(o for o in get_args(t) if o!=type(None))\n",
    "    d = {bool: str2bool, int: str2int}\n",
    "    res = d.get(t, t)\n",
    "    if origin in (list,List): return partial(_mk_list, res)\n",
    "    return lambda o: res(o[-1]) if isinstance(o,(list,tuple)) else res(o)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "95b1f5c9",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(_fix_anno(Union[str,None])('a'), 'a')\n",
    "test_eq(_fix_anno(float)(0.9), 0.9)\n",
    "test_eq(_fix_anno(int)('1'), 1)\n",
    "test_eq(_fix_anno(int)(['1','2']), 2)\n",
    "test_eq(_fix_anno(list[int])(['1','2']), [1,2])\n",
    "test_eq(_fix_anno(list[int])('1'), [1])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c58ccadb",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _form_arg(k, v, d):\n",
    "    \"Get type by accessing key `k` from `d`, and use to cast `v`\"\n",
    "    if v is None: return\n",
    "    if not isinstance(v, (str,list,tuple)): return v\n",
    "    # This is the type we want to cast `v` to\n",
    "    anno = d.get(k, None)\n",
    "    if not anno: return v\n",
    "    return _fix_anno(anno)(v)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "59757d76",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = dict(k=int, l=List[int])\n",
    "test_eq(_form_arg('k', \"1\", d), 1)\n",
    "test_eq(_form_arg('l', \"1\", d), [1])\n",
    "test_eq(_form_arg('l', [\"1\",\"2\"], d), [1,2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5fc04751",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@dataclass\n",
    "class HttpHeader: k:str;v:str"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "56cc589f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _annotations(anno):\n",
    "    \"Same as `get_annotations`, but also works on namedtuples\"\n",
    "    if is_namedtuple(anno): return {o:str for o in anno._fields}\n",
    "    return get_annotations(anno)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cb4ed4aa",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _is_body(anno): return issubclass(anno, (dict,ns)) or _annotations(anno)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "ffdde66f",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _formitem(form, k):\n",
    "    \"Return single item `k` from `form` if len 1, otherwise return list\"\n",
    "    o = form.getlist(k)\n",
    "    return o[0] if len(o) == 1 else o if o else None"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5fe74444",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def form2dict(form: FormData) -> dict:\n",
    "    \"Convert starlette form data to a dict\"\n",
    "    return {k: _formitem(form, k) for k in form}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cf80ea34",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = [('a',1),('a',2),('b',0)]\n",
    "fd = FormData(d)\n",
    "res = form2dict(fd)\n",
    "test_eq(res['a'], [1,2])\n",
    "test_eq(res['b'], 0)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b6ac588",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _from_body(req, p):\n",
    "    anno = p.annotation\n",
    "    # Get the fields and types of type `anno`, if available\n",
    "    d = _annotations(anno)\n",
    "    if req.headers.get('content-type', None)=='application/json': data = await req.json()\n",
    "    else: data = form2dict(await req.form())\n",
    "    cargs = {k: _form_arg(k, v, d) for k, v in data.items() if not d or k in d}\n",
    "    return anno(**cargs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2a8b10f4",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "{'k': 'value1', 'v': 'value3'}\n"
     ]
    }
   ],
   "source": [
    "async def f(req):\n",
    "    def _f(p:HttpHeader): ...\n",
    "    p = first(_sig(_f).parameters.values())\n",
    "    result = await _from_body(req, p)\n",
    "    return JSONResponse(result.__dict__)\n",
    "\n",
    "app = Starlette(routes=[Route('/', f, methods=['POST'])])\n",
    "client = TestClient(app)\n",
    "\n",
    "d = dict(k='value1',v=['value2','value3'])\n",
    "response = client.post('/', data=d)\n",
    "print(response.json())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6775cbf8",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _find_p(req, arg:str, p:Parameter):\n",
    "    \"In `req` find param named `arg` of type in `p` (`arg` is ignored for body types)\"\n",
    "    anno = p.annotation\n",
    "    # If there's an annotation of special types, return object of that type\n",
    "    # GenericAlias is a type of typing for iterators like list[int] that is not a class\n",
    "    if isinstance(anno, type) and not isinstance(anno, GenericAlias):\n",
    "        if issubclass(anno, Request): return req\n",
    "        if issubclass(anno, HtmxHeaders): return _get_htmx(req.headers)\n",
    "        if issubclass(anno, Starlette): return req.scope['app']\n",
    "        if _is_body(anno): return await _from_body(req, p)\n",
    "    # If there's no annotation, check for special names\n",
    "    if anno is empty:\n",
    "        if 'request'.startswith(arg.lower()): return req\n",
    "        if 'session'.startswith(arg.lower()): return req.scope.get('session', {})\n",
    "        if arg.lower()=='auth': return req.scope.get('auth', None)\n",
    "        if arg.lower()=='htmx': return _get_htmx(req.headers)\n",
    "        if arg.lower()=='app': return req.scope['app']\n",
    "        if arg.lower()=='body': return (await req.body()).decode()\n",
    "        if arg.lower() in ('hdrs','ftrs','bodykw','htmlkw'): return getattr(req, arg.lower())\n",
    "        if arg!='resp': warn(f\"`{arg} has no type annotation and is not a recognised special name, so is ignored.\")\n",
    "        return None\n",
    "    # Look through path, cookies, headers, query, and body in that order\n",
    "    res = req.path_params.get(arg, None)\n",
    "    if res in (empty,None): res = req.cookies.get(arg, None)\n",
    "    if res in (empty,None): res = req.headers.get(snake2hyphens(arg), None)\n",
    "    if res in (empty,None): res = req.query_params.get(arg, None)\n",
    "    if res in (empty,None):\n",
    "        frm = await req.form()\n",
    "        res = _formitem(frm, arg)\n",
    "    # Raise 400 error if the param does not include a default\n",
    "    if (res in (empty,None)) and p.default is empty: raise HTTPException(400, f\"Missing required field: {arg}\")\n",
    "    # If we have a default, return that if we have no value\n",
    "    if res in (empty,None): res = p.default\n",
    "    # We can cast str and list[str] to types; otherwise just return what we have\n",
    "    if not isinstance(res, (list,str)) or anno is empty: return res\n",
    "    anno = _fix_anno(anno)\n",
    "    try: return anno(res)\n",
    "    except ValueError: raise HTTPException(404, req.url.path) from None\n",
    "\n",
    "async def _wrap_req(req, params):\n",
    "    return [await _find_p(req, arg, p) for arg,p in params.items()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf945ee8",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[<starlette.requests.Request object>, <starlette.applications.Starlette object>, '1', HttpHeader(k='value1', v='value3')]\n"
     ]
    }
   ],
   "source": [
    "def g(req, this:Starlette, a:str, b:HttpHeader): ...\n",
    "\n",
    "async def f(req):\n",
    "    a = await _wrap_req(req, _sig(g).parameters)\n",
    "    return Response(str(a))\n",
    "\n",
    "app = Starlette(routes=[Route('/', f, methods=['POST'])])\n",
    "client = TestClient(app)\n",
    "\n",
    "response = client.post('/?a=1', data=d)\n",
    "print(response.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "7a661bfa",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def flat_xt(lst):\n",
    "    \"Flatten lists\"\n",
    "    result = []\n",
    "    if isinstance(lst,(FT,str)): lst=[lst]\n",
    "    for item in lst:\n",
    "        if isinstance(item, (list,tuple)): result.extend(item)\n",
    "        else: result.append(item)\n",
    "    return result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2ee5adf1",
   "metadata": {},
   "outputs": [],
   "source": [
    "x = ft('a',1)\n",
    "test_eq(flat_xt([x, x, [x,x]]), [x]*4)\n",
    "test_eq(flat_xt(x), [x])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "aacff5ac",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class Beforeware:\n",
    "    def __init__(self, f, skip=None): self.f,self.skip = f,skip or []"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "78c3c357",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _handle(f, args, **kwargs):\n",
    "    return (await f(*args, **kwargs)) if is_async_callable(f) else await run_in_threadpool(f, *args, **kwargs)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "079a3215",
   "metadata": {},
   "source": [
    "## Websockets"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f2277c02",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _find_wsp(ws, data, hdrs, arg:str, p:Parameter):\n",
    "    \"In `data` find param named `arg` of type in `p` (`arg` is ignored for body types)\"\n",
    "    anno = p.annotation\n",
    "    if isinstance(anno, type):\n",
    "        if issubclass(anno, HtmxHeaders): return _get_htmx(hdrs)\n",
    "        if issubclass(anno, Starlette): return ws.scope['app']\n",
    "    if anno is empty:\n",
    "        if arg.lower()=='ws': return ws\n",
    "        if arg.lower()=='data': return data\n",
    "        if arg.lower()=='htmx': return _get_htmx(hdrs)\n",
    "        if arg.lower()=='app': return ws.scope['app']\n",
    "        if arg.lower()=='send': return partial(_send_ws, ws)\n",
    "        return None\n",
    "    res = data.get(arg, None)\n",
    "    if res is empty or res is None: res = hdrs.get(snake2hyphens(arg), None)\n",
    "    if res is empty or res is None: res = p.default\n",
    "    # We can cast str and list[str] to types; otherwise just return what we have\n",
    "    if not isinstance(res, (list,str)) or anno is empty: return res\n",
    "    anno = _fix_anno(anno)\n",
    "    return [anno(o) for o in res] if isinstance(res,list) else anno(res)\n",
    "\n",
    "def _wrap_ws(ws, data, params):\n",
    "    hdrs = data.pop('HEADERS', {})\n",
    "    return [_find_wsp(ws, data, hdrs, arg, p) for arg,p in params.items()]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "dcc15129",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _send_ws(ws, resp):\n",
    "    if not resp: return\n",
    "    res = to_xml(resp, indent=fh_cfg.indent) if isinstance(resp, (list,tuple,FT)) or hasattr(resp, '__ft__') else resp\n",
    "    await ws.send_text(res)\n",
    "\n",
    "def _ws_endp(recv, conn=None, disconn=None, hdrs=None, before=None):\n",
    "    cls = type('WS_Endp', (WebSocketEndpoint,), {\"encoding\":\"text\"})\n",
    "    \n",
    "    async def _generic_handler(handler, ws, data=None):\n",
    "        wd = _wrap_ws(ws, loads(data) if data else {}, _sig(handler).parameters)\n",
    "        resp = await _handle(handler, wd)\n",
    "        if resp: await _send_ws(ws, resp)\n",
    "\n",
    "    async def _connect(self, ws):\n",
    "        await ws.accept()\n",
    "        await _generic_handler(conn, ws)\n",
    "\n",
    "    async def _disconnect(self, ws, close_code): await _generic_handler(disconn, ws)\n",
    "    async def _recv(self, ws, data): await _generic_handler(recv, ws, data)\n",
    "\n",
    "    if    conn: cls.on_connect    = _connect\n",
    "    if disconn: cls.on_disconnect = _disconnect\n",
    "    cls.on_receive = _recv\n",
    "    return cls"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "983bcfe2",
   "metadata": {},
   "outputs": [],
   "source": [
    "def on_receive(self, msg:str): return f\"Message text was: {msg}\"\n",
    "c = _ws_endp(on_receive)\n",
    "app = Starlette(routes=[WebSocketRoute('/', _ws_endp(on_receive))])\n",
    "\n",
    "cli = TestClient(app)\n",
    "with cli.websocket_connect('/') as ws:\n",
    "    ws.send_text('{\"msg\":\"Hi!\"}')\n",
    "    data = ws.receive_text()\n",
    "    assert data == 'Message text was: Hi!'"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "496cd78f",
   "metadata": {},
   "source": [
    "## Routing and application"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b89130cd",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class WS_RouteX(WebSocketRoute):\n",
    "    def __init__(self, path:str, recv, conn:callable=None, disconn:callable=None, *,\n",
    "                 name=None, middleware=None, hdrs=None, before=None):\n",
    "        super().__init__(path, _ws_endp(recv, conn, disconn, hdrs, before), name=name, middleware=middleware)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "03f4c639",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def uri(_arg, **kwargs):\n",
    "    return f\"{quote(_arg)}/{urlencode(kwargs, doseq=True)}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bdb2ac14",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def decode_uri(s): \n",
    "    arg,_,kw = s.partition('/')\n",
    "    return unquote(arg), {k:v[0] for k,v in parse_qs(kw).items()}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b80ce139",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "from starlette.convertors import StringConvertor"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a27adc1e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "StringConvertor.regex = \"[^/]*\"  # `+` replaced with `*`\n",
    "\n",
    "@patch\n",
    "def to_string(self:StringConvertor, value: str) -> str:\n",
    "    value = str(value)\n",
    "    assert \"/\" not in value, \"May not contain path separators\"\n",
    "    # assert value, \"Must not be empty\"  # line removed due to errors\n",
    "    return value"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "46614165",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "@patch\n",
    "def url_path_for(self:HTTPConnection, name: str, **path_params):\n",
    "    lp = self.scope['app'].url_path_for(name, **path_params)\n",
    "    return URLPath(f\"{self.scope['root_path']}{lp}\", lp.protocol, lp.host)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c1707d59",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "_verbs = dict(get='hx-get', post='hx-post', put='hx-post', delete='hx-delete', patch='hx-patch', link='href')\n",
    "\n",
    "def _url_for(req, t):\n",
    "    if callable(t): t = t.__routename__\n",
    "    kw = {}\n",
    "    if t.find('/')>-1 and (t.find('?')<0 or t.find('/')<t.find('?')): t,kw = decode_uri(t)\n",
    "    t,m,q = t.partition('?')    \n",
    "    return f\"{req.url_path_for(t, **kw)}{m}{q}\"\n",
    "\n",
    "def _find_targets(req, resp):\n",
    "    if isinstance(resp, tuple):\n",
    "        for o in resp: _find_targets(req, o)\n",
    "    if isinstance(resp, FT):\n",
    "        for o in resp.children: _find_targets(req, o)\n",
    "        for k,v in _verbs.items():\n",
    "            t = resp.attrs.pop(k, None)\n",
    "            if t: resp.attrs[v] = _url_for(req, t)\n",
    "\n",
    "def _apply_ft(o):\n",
    "    if isinstance(o, tuple): o = tuple(_apply_ft(c) for c in o)\n",
    "    if hasattr(o, '__ft__'): o = o.__ft__()\n",
    "    if isinstance(o, FT): o.children = [_apply_ft(c) for c in o.children]\n",
    "    return o\n",
    "\n",
    "def _to_xml(req, resp, indent):\n",
    "    resp = _apply_ft(resp)\n",
    "    _find_targets(req, resp)\n",
    "    return to_xml(resp, indent)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a407dc0e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def flat_tuple(o):\n",
    "    \"Flatten lists\"\n",
    "    result = []\n",
    "    if not isinstance(o,(tuple,list)): o=[o]\n",
    "    o = list(o)\n",
    "    for item in o:\n",
    "        if isinstance(item, (list,tuple)): result.extend(item)\n",
    "        else: result.append(item)\n",
    "    return tuple(result)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4095049d",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _xt_resp(req, resp):\n",
    "    resp = flat_tuple(resp)\n",
    "    resp = resp + tuple(getattr(req, 'injects', ()))\n",
    "    http_hdrs,resp = partition(resp, risinstance(HttpHeader))\n",
    "    http_hdrs = {o.k:str(o.v) for o in http_hdrs}\n",
    "    hdr_tags = 'title','meta','link','style','base'\n",
    "    titles,bdy = partition(resp, lambda o: getattr(o, 'tag', '') in hdr_tags)\n",
    "    if resp and 'hx-request' not in req.headers and not any(getattr(o, 'tag', '')=='html' for o in resp):\n",
    "        if not titles: titles = [Title('FastHTML page')]\n",
    "        resp = Html(Head(*titles, *flat_xt(req.hdrs)), Body(bdy, *flat_xt(req.ftrs), **req.bodykw), **req.htmlkw)\n",
    "    return HTMLResponse(_to_xml(req, resp, indent=fh_cfg.indent), headers=http_hdrs)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "968d9245",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _resp(req, resp, cls=empty):\n",
    "    if not resp: resp=()\n",
    "    if cls in (Any,FT): cls=empty\n",
    "    if isinstance(resp, FileResponse) and not os.path.exists(resp.path): raise HTTPException(404, resp.path)\n",
    "    if isinstance(resp, Response): return resp\n",
    "    if cls is not empty: return cls(resp)\n",
    "    if isinstance(resp, (list,tuple,HttpHeader,FT)) or hasattr(resp, '__ft__'): return _xt_resp(req, resp)\n",
    "    if isinstance(resp, str): cls = HTMLResponse\n",
    "    elif isinstance(resp, Mapping): cls = JSONResponse\n",
    "    else:\n",
    "        resp = str(resp)\n",
    "        cls = HTMLResponse\n",
    "    return cls(resp)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1ba52822",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "async def _wrap_call(f, req, params):\n",
    "    wreq = await _wrap_req(req, params)\n",
    "    return await _handle(f, wreq)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "541cfa0a",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class RouteX(Route):\n",
    "    def __init__(self, path:str, endpoint, *, methods=None, name=None, include_in_schema=True, middleware=None,\n",
    "                hdrs=None, ftrs=None, before=None, after=None, htmlkw=None, **bodykw):\n",
    "        self.sig = _sig(endpoint)\n",
    "        self.f,self.hdrs,self.ftrs,self.before,self.after,self.htmlkw,self.bodykw = endpoint,hdrs,ftrs,before,after,htmlkw,bodykw\n",
    "        super().__init__(path, self._endp, methods=methods, name=name, include_in_schema=include_in_schema, middleware=middleware)\n",
    "\n",
    "    async def _endp(self, req):\n",
    "        resp = None\n",
    "        req.injects = []\n",
    "        req.hdrs,req.ftrs,req.htmlkw,req.bodykw = map(deepcopy, (self.hdrs,self.ftrs,self.htmlkw,self.bodykw))\n",
    "        req.hdrs,req.ftrs = list(req.hdrs),list(req.ftrs)\n",
    "        for b in self.before:\n",
    "            if not resp:\n",
    "                if isinstance(b, Beforeware): bf,skip = b.f,b.skip\n",
    "                else: bf,skip = b,[]\n",
    "                if not any(re.fullmatch(r, req.url.path) for r in skip):\n",
    "                    resp = await _wrap_call(bf, req, _sig(bf).parameters)\n",
    "        if not resp: resp = await _wrap_call(self.f, req, self.sig.parameters)\n",
    "        for a in self.after:\n",
    "            _,*wreq = await _wrap_req(req, _sig(a).parameters)\n",
    "            nr = a(resp, *wreq)\n",
    "            if nr: resp = nr\n",
    "        return _resp(req, resp, self.sig.return_annotation)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e83fc876",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class RouterX(Router):\n",
    "    def __init__(self, routes=None, redirect_slashes=True, default=None, on_startup=None, on_shutdown=None,\n",
    "                 lifespan=None, *, middleware=None, hdrs=None, ftrs=None, before=None, after=None, htmlkw=None, **bodykw):\n",
    "        super().__init__(routes, redirect_slashes, default, on_startup, on_shutdown,\n",
    "                 lifespan=lifespan, middleware=middleware)\n",
    "        self.hdrs,self.ftrs,self.bodykw,self.htmlkw,self.before,self.after = hdrs,ftrs,bodykw,htmlkw or {},before,after\n",
    "\n",
    "    def add_route( self, path: str, endpoint: callable, methods=None, name=None, include_in_schema=True):\n",
    "        route = RouteX(path, endpoint=endpoint, methods=methods, name=name, include_in_schema=include_in_schema,\n",
    "                       hdrs=self.hdrs, ftrs=self.ftrs, before=self.before, after=self.after, htmlkw=self.htmlkw, **self.bodykw)\n",
    "        self.routes.append(route)\n",
    "\n",
    "    def add_ws( self, path: str, recv: callable, conn:callable=None, disconn:callable=None, name=None):\n",
    "        route = WS_RouteX(path, recv=recv, conn=conn, disconn=disconn, name=name, hdrs=self.hdrs, before=self.before)\n",
    "        self.routes.append(route)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fb97d46b",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "htmxsrc   = Script(src=\"https://unpkg.com/htmx.org@next/dist/htmx.min.js\")\n",
    "htmxwssrc = Script(src=\"https://unpkg.com/htmx-ext-ws/ws.js\")\n",
    "fhjsscr   = Script(src=\"https://cdn.jsdelivr.net/gh/answerdotai/fasthtml-js@main/fasthtml.js\")\n",
    "htmxctsrc = Script(src=\"https://unpkg.com/htmx-ext-transfer-encoding-chunked/transfer-encoding-chunked.js\")\n",
    "surrsrc   = Script(src=\"https://cdn.jsdelivr.net/gh/answerdotai/surreal@main/surreal.js\")\n",
    "scopesrc  = Script(src=\"https://cdn.jsdelivr.net/gh/gnat/css-scope-inline@main/script.js\")\n",
    "viewport  = Meta(name=\"viewport\", content=\"width=device-width, initial-scale=1, viewport-fit=cover\")\n",
    "charset   = Meta(charset=\"utf-8\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8bd78eeb",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def get_key(key=None, fname='.sesskey'):\n",
    "    if key: return key\n",
    "    fname = Path(fname)\n",
    "    if fname.exists(): return fname.read_text()\n",
    "    key = str(uuid.uuid4())\n",
    "    fname.write_text(key)\n",
    "    return key"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4771838e",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'480ec94e-c759-4105-afd8-3a64af020118'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "get_key()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "042666e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _list(o): return [] if not o else list(o) if isinstance(o, (tuple,list)) else [o]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d276fc71",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _wrap_ex(f, hdrs, ftrs, htmlkw, bodykw):\n",
    "    async def _f(req, exc):\n",
    "        req.hdrs,req.ftrs,req.htmlkw,req.bodykw = map(deepcopy, (hdrs, ftrs, htmlkw, bodykw))\n",
    "        res = await _handle(f, (req, exc))\n",
    "        return _resp(req, res)\n",
    "    return _f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "00a0896c",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def _mk_locfunc(f,p):\n",
    "    class _lf:\n",
    "        def __init__(self): update_wrapper(self, f)\n",
    "        def __call__(self, *args, **kw): return f(*args, **kw)\n",
    "        def rt(self, **kw): return p + (f'?{urlencode(kw)}' if kw else '')\n",
    "        def __str__(self): return p\n",
    "    return _lf()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d254febe",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class FastHTML(Starlette):\n",
    "    def __init__(self, debug=False, routes=None, middleware=None, exception_handlers=None,\n",
    "                 on_startup=None, on_shutdown=None, lifespan=None, hdrs=None, ftrs=None,\n",
    "                 before=None, after=None, ws_hdr=False, ct_hdr=False,\n",
    "                 surreal=True, htmx=True, default_hdrs=True, sess_cls=SessionMiddleware,\n",
    "                 secret_key=None, session_cookie='session_', max_age=365*24*3600, sess_path='/',\n",
    "                 same_site='lax', sess_https_only=False, sess_domain=None, key_fname='.sesskey',\n",
    "                 htmlkw=None, **bodykw):\n",
    "        middleware,before,after = map(_list, (middleware,before,after))\n",
    "        secret_key = get_key(secret_key, key_fname)\n",
    "        if sess_cls:\n",
    "            sess = Middleware(sess_cls, secret_key=secret_key,session_cookie=session_cookie,\n",
    "                              max_age=max_age, path=sess_path, same_site=same_site,\n",
    "                              https_only=sess_https_only, domain=sess_domain)\n",
    "            middleware.append(sess)\n",
    "        hdrs,ftrs = listify(hdrs),listify(ftrs)\n",
    "        htmlkw = htmlkw or {}\n",
    "        if default_hdrs:\n",
    "            if surreal: hdrs = [surrsrc,scopesrc] + hdrs\n",
    "            if ws_hdr: hdrs = [htmxwssrc] + hdrs\n",
    "            if ct_hdr: hdrs = [htmxctsrc] + hdrs\n",
    "            if htmx: hdrs = [htmxsrc,fhjsscr] + hdrs\n",
    "            hdrs = [charset, viewport] + hdrs\n",
    "        exception_handlers = ifnone(exception_handlers, {})\n",
    "        if 404 not in exception_handlers: \n",
    "            def _not_found(req, exc): return  Response('404 Not Found', status_code=404)  \n",
    "            exception_handlers[404] = _not_found\n",
    "        excs = {k:_wrap_ex(v, hdrs, ftrs, htmlkw, bodykw) for k,v in exception_handlers.items()}\n",
    "        super().__init__(debug, routes, middleware, excs, on_startup, on_shutdown, lifespan=lifespan)\n",
    "        self.router = RouterX(routes, on_startup=on_startup, on_shutdown=on_shutdown, lifespan=lifespan,\n",
    "                              hdrs=hdrs, ftrs=ftrs, before=before, after=after, htmlkw=htmlkw, **bodykw)\n",
    "\n",
    "    def ws(self, path:str, conn=None, disconn=None, name=None):\n",
    "        \"Add a websocket route at `path`\"\n",
    "        def f(func):\n",
    "            self.router.add_ws(path, func, conn=conn, disconn=disconn, name=name)\n",
    "            return func\n",
    "        return f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "72760b09",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "all_meths = 'get post put delete patch head trace options'.split()\n",
    "\n",
    "@patch\n",
    "def route(self:FastHTML, path:str=None, methods=None, name=None, include_in_schema=True):\n",
    "    \"Add a route at `path`\"\n",
    "    pathstr = None if callable(path) else path\n",
    "    def f(func):\n",
    "        n,fn,p = name,func.__name__,pathstr\n",
    "        assert path or (fn not in all_meths), \"Must provide a path when using http verb-based function name\"\n",
    "        if methods: m = [methods] if isinstance(methods,str) else methods\n",
    "        else: m = [fn] if fn in all_meths else ['get','post']\n",
    "        if not n: n = fn\n",
    "        if not p: p = '/'+('' if fn=='index' else fn)\n",
    "        self.router.add_route(p, func, methods=m, name=n, include_in_schema=include_in_schema)\n",
    "        lf = _mk_locfunc(func, p)\n",
    "        lf.__routename__ = n\n",
    "        return lf\n",
    "    return f(path) if callable(path) else f"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3127c92e",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "for o in all_meths: setattr(FastHTML, o, partialmethod(FastHTML.route, methods=o))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3a348474",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def serve(\n",
    "        appname=None, # Name of the module\n",
    "        app='app', # App instance to be served\n",
    "        host='0.0.0.0', # If host is 0.0.0.0 will convert to localhost\n",
    "        port=None, # If port is None it will default to 5001 or the PORT environment variable\n",
    "        reload=True, # Default is to reload the app upon code changes\n",
    "        reload_includes:list[str]|str|None=None, # Additional files to watch for changes\n",
    "        reload_excludes:list[str]|str|None=None # Files to ignore for changes\n",
    "        ): \n",
    "    \"Run the app in an async server, with live reload set as the default.\"\n",
    "    bk = inspect.currentframe().f_back\n",
    "    glb = bk.f_globals\n",
    "    code = bk.f_code\n",
    "    if not appname:\n",
    "        if glb.get('__name__')=='__main__': appname = Path(glb.get('__file__', '')).stem\n",
    "        elif code.co_name=='main' and bk.f_back.f_globals.get('__name__')=='__main__': appname = inspect.getmodule(bk).__name__\n",
    "    if appname:\n",
    "        if not port: port=int(os.getenv(\"PORT\", default=5001))\n",
    "        print(f'Link: http://{\"localhost\" if host==\"0.0.0.0\" else host}:{port}')\n",
    "        uvicorn.run(f'{appname}:{app}', host=host, port=port, reload=reload, reload_includes=reload_includes, reload_excludes=reload_excludes)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "374d5e24",
   "metadata": {},
   "source": [
    "## Extras"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "402ec641",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def cookie(key: str, value=\"\", max_age=None, expires=None, path=\"/\", domain=None, secure=False, httponly=False, samesite=\"lax\",):\n",
    "    \"Create a 'set-cookie' `HttpHeader`\"\n",
    "    cookie = cookies.SimpleCookie()\n",
    "    cookie[key] = value\n",
    "    if max_age is not None: cookie[key][\"max-age\"] = max_age\n",
    "    if expires is not None:\n",
    "        cookie[key][\"expires\"] = format_datetime(expires, usegmt=True) if isinstance(expires, datetime) else expires\n",
    "    if path is not None: cookie[key][\"path\"] = path\n",
    "    if domain is not None: cookie[key][\"domain\"] = domain\n",
    "    if secure: cookie[key][\"secure\"] = True\n",
    "    if httponly: cookie[key][\"httponly\"] = True\n",
    "    if samesite is not None:\n",
    "        assert samesite.lower() in [ \"strict\", \"lax\", \"none\", ], \"must be 'strict', 'lax' or 'none'\"\n",
    "        cookie[key][\"samesite\"] = samesite\n",
    "    cookie_val = cookie.output(header=\"\").strip()\n",
    "    return HttpHeader(\"set-cookie\", cookie_val)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "91c223f9",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "def reg_re_param(m, s):\n",
    "    cls = get_class(f'{m}Conv', sup=StringConvertor, regex=s)\n",
    "    register_url_convertor(m, cls())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3d252538",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "# Starlette doesn't have the '?', so it chomps the whole remaining URL\n",
    "reg_re_param(\"path\", \".*?\")\n",
    "reg_re_param(\"static\", \"ico|gif|jpg|jpeg|webm|css|js|woff|png|svg|mp4|webp|ttf|otf|eot|woff2|txt|html|map\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0b727f38",
   "metadata": {},
   "outputs": [],
   "source": [
    "#| export\n",
    "class MiddlewareBase:\n",
    "    async def __call__(self, scope, receive, send) -> None:\n",
    "        if scope[\"type\"] not in [\"http\", \"websocket\"]:\n",
    "            await self.app(scope, receive, send)\n",
    "            return\n",
    "        return HTTPConnection(scope)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2d2633dd",
   "metadata": {},
   "source": [
    "## Tests"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9abc3781",
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_cli(app): return app,TestClient(app),app.route"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "645d8d95",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML(secret_key='soopersecret'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4d7773d1",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Hi there'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/hi\")\n",
    "def get(): return 'Hi there'\n",
    "\n",
    "r = cli.get('/hi')\n",
    "r.text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a8e723fc",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Postal'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/hi\")\n",
    "def post(): return 'Postal'\n",
    "\n",
    "cli.post('/hi').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "274666db",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'testserver'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@app.get(\"/hostie\")\n",
    "def show_host(req): return req.headers['host']\n",
    "\n",
    "cli.get('/hostie').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "36292bf3",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'a yoyo'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt\n",
    "def yoyo(): return 'a yoyo'\n",
    "\n",
    "cli.post('/yoyo').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0d813cd0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<!doctype html>\n",
      "\n",
      "<html><div hx-post=\"a yoyo\">Text.</div>\n",
      "</html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autopost(): return Html(Div('Text.', hx_post=yoyo()))\n",
    "print(cli.get('/autopost').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52dde0da",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<!doctype html>\n",
      "\n",
      "<html><body><div class=\"px-2\" hx-post=\"/hostie?a=b\">Text.</div>\n",
      "</body>\n",
      "</html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autopost2(): return Html(Body(Div('Text.', cls='px-2', hx_post=show_host.rt(a='b'))))\n",
    "print(cli.get('/autopost2').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d84d98f1",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<!doctype html>\n",
      "\n",
      "<html><div hx-get=\"/hostie\">Text.</div>\n",
      "</html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autoget2(): return Html(Div('Text.', hx_get=show_host))\n",
    "print(cli.get('/autoget2').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "64343367",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'Good day to you, Alexis!'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt('/user/{nm}', name='gday')\n",
    "def get(nm:str=''): return f\"Good day to you, {nm}!\"\n",
    "cli.get('/user/Alexis').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "29a96715",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<!doctype html>\n",
      "\n",
      "<html><div href=\"/user/Alexis\">Text.</div>\n",
      "</html>\n",
      "\n"
     ]
    }
   ],
   "source": [
    "@app.get\n",
    "def autolink(): return Html(Div('Text.', link=uri('gday', nm='Alexis')))\n",
    "print(cli.get('/autolink').text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "54266599",
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'http://testserver/user/Alexis; http://testserver/hostie'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt('/link')\n",
    "def get(req): return f\"{req.url_for('gday', nm='Alexis')}; {req.url_for('show_host')}\"\n",
    "\n",
    "cli.get('/link').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "db0281cf",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_eq(app.router.url_path_for('gday', nm='Jeremy'), '/user/Jeremy')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b827960a",
   "metadata": {},
   "outputs": [],
   "source": [
    "hxhdr = {'headers':{'hx-request':\"1\"}}\n",
    "\n",
    "@rt('/ft')\n",
    "def get(): return Title('Foo'),H1('bar')\n",
    "\n",
    "txt = cli.get('/ft').text\n",
    "assert '<title>Foo</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt\n",
    "\n",
    "@rt('/xt2')\n",
    "def get(): return H1('bar')\n",
    "\n",
    "txt = cli.get('/xt2').text\n",
    "assert '<title>FastHTML page</title>' in txt and '<h1>bar</h1>' in txt and '<html>' in txt\n",
    "\n",
    "assert cli.get('/xt2', **hxhdr).text.strip() == '<h1>bar</h1>'\n",
    "\n",
    "@rt('/xt3')\n",
    "def get(): return Html(Head(Title('hi')), Body(P('there')))\n",
    "\n",
    "txt = cli.get('/xt3').text\n",
    "assert '<title>FastHTML page</title>' not in txt and '<title>hi</title>' in txt and '<p>there</p>' in txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "381a1b68",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt('/oops')\n",
    "def get(nope): return nope\n",
    "test_warns(lambda: cli.get('/oops?nope=1'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6b022adb",
   "metadata": {},
   "outputs": [],
   "source": [
    "def test_r(cli, path, exp, meth='get', hx=False, **kwargs):\n",
    "    if hx: kwargs['headers'] = {'hx-request':\"1\"}\n",
    "    test_eq(getattr(cli, meth)(path, **kwargs).text, exp)\n",
    "\n",
    "app.chk = 'foo'\n",
    "ModelName = str_enum('ModelName', \"alexnet\", \"resnet\", \"lenet\")\n",
    "fake_db = [{\"name\": \"Foo\"}, {\"name\": \"Bar\"}]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "0bfd7af6",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt('/html/{idx}')\n",
    "async def get(idx:int): return Body(H4(f'Next is {idx+1}.'))\n",
    "\n",
    "reg_re_param(\"imgext\", \"ico|gif|jpg|jpeg|webm\")\n",
    "\n",
    "@rt(r'/static/{path:path}{fn}.{ext:imgext}')\n",
    "def get(fn:str, path:str, ext:str): return f\"Getting {fn}.{ext} from /{path}\"\n",
    "\n",
    "@rt(\"/models/{nm}\")\n",
    "def get(nm:ModelName): return nm\n",
    "\n",
    "@rt(\"/files/{path}\")\n",
    "async def get(path: Path): return path.with_suffix('.txt')\n",
    "\n",
    "@rt(\"/items/\")\n",
    "def get(idx:int|None = 0): return fake_db[idx]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52d9a3f2",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_r(cli, '/html/1', '<body><h4>Next is 2.</h4>\\n</body>\\n', hx=True)\n",
    "test_r(cli, '/static/foo/jph.ico', 'Getting jph.ico from /foo/')\n",
    "test_r(cli, '/models/alexnet', 'alexnet')\n",
    "test_r(cli, '/files/foo', 'foo.txt')\n",
    "test_r(cli, '/items/?idx=1', '{\"name\":\"Bar\"}')\n",
    "test_r(cli, '/items/', '{\"name\":\"Foo\"}')\n",
    "assert cli.get('/items/?idx=g').text=='404 Not Found'\n",
    "assert cli.get('/items/?idx=g').status_code == 404"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b8be4ef3",
   "metadata": {},
   "outputs": [],
   "source": [
    "@app.get(\"/booly/\")\n",
    "def _(coming:bool=True): return 'Coming' if coming else 'Not coming'\n",
    "\n",
    "@app.get(\"/datie/\")\n",
    "def _(d:date): return d\n",
    "\n",
    "@app.get(\"/ua\")\n",
    "async def _(user_agent:str): return user_agent\n",
    "\n",
    "@app.get(\"/hxtest\")\n",
    "def _(htmx): return htmx.request\n",
    "\n",
    "@app.get(\"/hxtest2\")\n",
    "def _(foo:HtmxHeaders, req): return foo.request\n",
    "\n",
    "@app.get(\"/app\")\n",
    "def _(app): return app.chk\n",
    "\n",
    "@app.get(\"/app2\")\n",
    "def _(foo:FastHTML): return foo.chk,HttpHeader(\"mykey\", \"myval\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5f045cf9",
   "metadata": {},
   "outputs": [],
   "source": [
    "test_r(cli, '/booly/?coming=true', 'Coming')\n",
    "test_r(cli, '/booly/?coming=no', 'Not coming')\n",
    "date_str = \"17th of May, 2024, 2p\"\n",
    "test_r(cli, f'/datie/?d={date_str}', '2024-05-17 14:00:00')\n",
    "test_r(cli, '/ua', 'FastHTML', headers={'User-Agent':'FastHTML'})\n",
    "test_r(cli, '/hxtest' , '1', headers={'HX-Request':'1'})\n",
    "test_r(cli, '/hxtest2', '1', headers={'HX-Request':'1'})\n",
    "test_r(cli, '/app' , 'foo')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "52f9f934",
   "metadata": {},
   "outputs": [],
   "source": [
    "r = cli.get('/app2', **hxhdr)\n",
    "test_eq(r.text, 'foo\\n')\n",
    "test_eq(r.headers['mykey'], 'myval')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a91b0af",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt\n",
    "def meta(): \n",
    "    return ((Title('hi'),H1('hi')),\n",
    "        (Meta(property='image'), Meta(property='site_name'))\n",
    "   )\n",
    "\n",
    "t = cli.post('/meta').text\n",
    "assert re.search(r'<body>\\s*<h1>hi</h1>\\s*</body>', t)\n",
    "assert '<meta' in t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a3596991",
   "metadata": {},
   "outputs": [],
   "source": [
    "@app.post('/profile/me')\n",
    "def profile_update(username: str): return username\n",
    "\n",
    "test_r(cli, '/profile/me', 'Alexis', 'post', data={'username' : 'Alexis'})\n",
    "test_r(cli, '/profile/me', 'Missing required field: username', 'post', data={})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fdb9239c",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Example post request with parameter that has a default value\n",
    "@app.post('/pet/dog')\n",
    "def pet_dog(dogname: str = None): return dogname\n",
    "\n",
    "# Working post request with optional parameter\n",
    "test_r(cli, '/pet/dog', '', 'post', data={})"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "3366d88f",
   "metadata": {},
   "outputs": [],
   "source": [
    "@dataclass\n",
    "class Bodie: a:int;b:str\n",
    "\n",
    "@rt(\"/bodie/{nm}\")\n",
    "def post(nm:str, data:Bodie):\n",
    "    res = asdict(data)\n",
    "    res['nm'] = nm\n",
    "    return res\n",
    "\n",
    "@app.post(\"/bodied/\")\n",
    "def bodied(data:dict): return data\n",
    "\n",
    "nt = namedtuple('Bodient', ['a','b'])\n",
    "\n",
    "@app.post(\"/bodient/\")\n",
    "def bodient(data:nt): return asdict(data)\n",
    "\n",
    "class BodieTD(TypedDict): a:int;b:str='foo'\n",
    "\n",
    "@app.post(\"/bodietd/\")\n",
    "def bodient(data:BodieTD): return data\n",
    "\n",
    "class Bodie2:\n",
    "    a:int|None; b:str\n",
    "    def __init__(self, a, b='foo'): store_attr()\n",
    "\n",
    "@app.post(\"/bodie2/\")\n",
    "def bodie(d:Bodie2): return f\"a: {d.a}; b: {d.b}\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fcc024e6",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.xtend import Titled"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6c1fbc4b",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Testing POST with Content-Type: application/json\n",
    "@app.post(\"/\")\n",
    "def index(it: Bodie): return Titled(\"It worked!\", P(f\"{it.a}, {it.b}\"))\n",
    "\n",
    "s = json.dumps({\"b\": \"Lorem\", \"a\": 15})\n",
    "response = cli.post('/', headers={\"Content-Type\": \"application/json\"}, data=s).text\n",
    "assert \"<title>It worked!</title>\" in response and \"<p>15, Lorem</p>\" in response"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "51cd4643",
   "metadata": {},
   "outputs": [],
   "source": [
    "# Testing POST with Content-Type: application/json\n",
    "@app.post(\"/bodytext\")\n",
    "def index(body): return body\n",
    "\n",
    "response = cli.post('/bodytext', headers={\"Content-Type\": \"application/json\"}, data=s).text\n",
    "test_eq(response, '{\"b\": \"Lorem\", \"a\": 15}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "48f0a45e",
   "metadata": {},
   "outputs": [],
   "source": [
    "d = dict(a=1, b='foo')\n",
    "\n",
    "test_r(cli, '/bodie/me', '{\"a\":1,\"b\":\"foo\",\"nm\":\"me\"}', 'post', data=dict(a=1, b='foo', nm='me'))\n",
    "test_r(cli, '/bodied/', '{\"a\":\"1\",\"b\":\"foo\"}', 'post', data=d)\n",
    "test_r(cli, '/bodie2/', 'a: 1; b: foo', 'post', data={'a':1})\n",
    "test_r(cli, '/bodient/', '{\"a\":\"1\",\"b\":\"foo\"}', 'post', data=d)\n",
    "test_r(cli, '/bodietd/', '{\"a\":1,\"b\":\"foo\"}', 'post', data=d)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8057b34b",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'Cookie was set at time 19:10:33.000439'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/setcookie\")\n",
    "def get(req): return cookie('now', datetime.now())\n",
    "\n",
    "@rt(\"/getcookie\")\n",
    "def get(now:date): return f'Cookie was set at time {now.time()}'\n",
    "\n",
    "print(cli.get('/setcookie').text)\n",
    "time.sleep(0.01)\n",
    "cli.get('/getcookie').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5bce33f0",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Set to 2024-08-25 19:10:33.039493\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'Session time: 2024-08-25 19:10:33.039493'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "@rt(\"/setsess\")\n",
    "def get(sess, foo:str=''):\n",
    "    now = datetime.now()\n",
    "    sess['auth'] = str(now)\n",
    "    return f'Set to {now}'\n",
    "\n",
    "@rt(\"/getsess\")\n",
    "def get(sess): return f'Session time: {sess[\"auth\"]}'\n",
    "\n",
    "print(cli.get('/setsess').text)\n",
    "time.sleep(0.01)\n",
    "\n",
    "cli.get('/getsess').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "249fce31",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/sess-first\")\n",
    "def post(sess, name: str):\n",
    "    sess[\"name\"] = name\n",
    "    return str(sess)\n",
    "\n",
    "cli.post('/sess-first', data={'name': 2})\n",
    "\n",
    "@rt(\"/getsess-all\")\n",
    "def get(sess): return sess['name']\n",
    "\n",
    "test_eq(cli.get('/getsess-all').text, '2')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4d17ab9c",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "# Release notes\n"
     ]
    }
   ],
   "source": [
    "@rt(\"/upload\")\n",
    "async def post(uf:UploadFile): return (await uf.read()).decode()\n",
    "\n",
    "with open('../../CHANGELOG.md', 'rb') as f:\n",
    "    print(cli.post('/upload', files={'uf':f}, data={'msg':'Hello'}).text[:15])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "9c1863bc",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/{fname:path}.{ext:static}\")\n",
    "async def get(fname:str, ext:str): return FileResponse(f'{fname}.{ext}')\n",
    "\n",
    "assert 'These are the source notebooks for FastHTML' in cli.get('/README.txt').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "102c17ab",
   "metadata": {},
   "outputs": [],
   "source": [
    "@rt(\"/form-submit/{list_id}\")\n",
    "def options(list_id: str):\n",
    "    headers = {\n",
    "        'Access-Control-Allow-Origin': '*',\n",
    "        'Access-Control-Allow-Methods': 'POST',\n",
    "        'Access-Control-Allow-Headers': '*',\n",
    "    }\n",
    "    return Response(status_code=200, headers=headers)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "85bf2984",
   "metadata": {},
   "outputs": [],
   "source": [
    "h = cli.options('/form-submit/2').headers\n",
    "test_eq(h['Access-Control-Allow-Methods'], 'POST')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e4fc13b0",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.authmw import user_pwd_auth"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "1f11eab1",
   "metadata": {},
   "outputs": [],
   "source": [
    "def _not_found(req, exc): return Div('nope')\n",
    "\n",
    "app,cli,rt = get_cli(FastHTML(exception_handlers={404:_not_found}))\n",
    "\n",
    "txt = cli.get('/').text\n",
    "assert '<div>nope</div>' in txt\n",
    "assert '<!doctype html>' in txt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "063b7f43",
   "metadata": {},
   "outputs": [],
   "source": [
    "app,cli,rt = get_cli(FastHTML())\n",
    "\n",
    "@rt(\"/{name}/{age}\")\n",
    "def get(name: str, age: int): \n",
    "    return Titled(f\"Hello {name.title()}, age {age}\")\n",
    "\n",
    "assert '<title>Hello Uma, age 5</title>' in cli.get('/uma/5').text\n",
    "assert '404 Not Found' in cli.get('/uma/five').text"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "efd6fc6c",
   "metadata": {},
   "outputs": [],
   "source": [
    "auth = user_pwd_auth(testuser='spycraft')\n",
    "app,cli,rt = get_cli(FastHTML(middleware=[auth]))\n",
    "\n",
    "@rt(\"/locked\")\n",
    "def get(auth): return 'Hello, ' + auth\n",
    "\n",
    "test_eq(cli.get('/locked').text, 'not authenticated')\n",
    "test_eq(cli.get('/locked', auth=(\"testuser\",\"spycraft\")).text, 'Hello, testuser')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "32520197",
   "metadata": {},
   "outputs": [],
   "source": [
    "auth = user_pwd_auth(testuser='spycraft')\n",
    "app,cli,rt = get_cli(FastHTML(middleware=[auth]))\n",
    "\n",
    "@rt(\"/locked\")\n",
    "def get(auth): return 'Hello, ' + auth\n",
    "\n",
    "test_eq(cli.get('/locked').text, 'not authenticated')\n",
    "test_eq(cli.get('/locked', auth=(\"testuser\",\"spycraft\")).text, 'Hello, testuser')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f26a6377",
   "metadata": {},
   "outputs": [],
   "source": [
    "from fasthtml.live_reload import FastHTMLWithLiveReload"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "a01d79b1",
   "metadata": {},
   "outputs": [],
   "source": [
    "hdrs, routes = app.router.hdrs, app.routes\n",
    "app,cli,rt = get_cli(FastHTMLWithLiveReload())\n",
    "\n",
    "@rt(\"/hi\")\n",
    "def get(): return 'Hi there'\n",
    "\n",
    "test_eq(cli.get('/hi').text, \"Hi there\")\n",
    "\n",
    "lr_hdrs, lr_routes = app.router.hdrs, app.routes\n",
    "test_eq(len(lr_hdrs), len(hdrs)+1)\n",
    "assert app.LIVE_RELOAD_HEADER in lr_hdrs\n",
    "test_eq(len(lr_routes), len(routes)+1)\n",
    "assert app.LIVE_RELOAD_ROUTE in lr_routes"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "474e14b4",
   "metadata": {},
   "source": [
    "## Export -"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d211e8e2",
   "metadata": {},
   "outputs": [],
   "source": [
    "#|hide\n",
    "import nbdev; nbdev.nbdev_export()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bf47445c",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
